package com.xian.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * 带拦截器的生产者
 * 即在发送数据前做一些处理
 */
public class InterceptorProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers","r71:9092");
        props.put("acks","all");
        props.put("retries",0);
        props.put("batch",16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");


        //构建拦截器
        List<String> interceptors = new ArrayList<String>();
        interceptors.add("com.xian.interceptor.TimeInterceptor"); //自定义拦截器的全类名
        interceptors.add("com.xian.interceptor.CountInterceptor"); //自定义拦截器的全类名

        //添加拦截器
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);



        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        //发送消息
        for(int i=1; i<=3; i++){
            producer.send(new ProducerRecord<String, String>("jemson-topic", i+"", "景甜" + i));
        }

        //关闭
        producer.close();



    }
}
